package cn.tpshion.pulsar.config;

import cn.tpshion.pulsar.support.ProducerFactoryBean;
import cn.tpshion.pulsar.support.PulsarProducer;
import cn.tpshion.pulsar.support.PulsarProperties;
import org.apache.pulsar.client.api.PulsarClient;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.List;

public class PulsarProducerConfiguration implements ApplicationContextAware {

    private PulsarProperties properties;

    private PulsarClient pulsarClient;

    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    public PulsarProducerConfiguration(PulsarProperties properties, PulsarClient pulsarClient) {
        this.properties = properties;
        this.pulsarClient = pulsarClient;
    }

    @PostConstruct
    public void createProducer(){
        List<PulsarProducer> producers = properties.getProducers();
        if(CollectionUtils.isEmpty(producers)){
            return;
        }
        DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory();
        for (PulsarProducer producer : producers) {
            BeanDefinitionBuilder definitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ProducerFactoryBean.class);
            definitionBuilder.addConstructorArgValue(pulsarClient);
            definitionBuilder.addConstructorArgValue(producer);
            AbstractBeanDefinition beanDefinition = definitionBuilder.getBeanDefinition();
            defaultListableBeanFactory.registerBeanDefinition(producer.getProducerName(), beanDefinition);
        }
    }

}
